package org.example.task;

import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

/**
 * 监听指定主题下的分组消息
 * topic 监听的主题 selectorExpression 监听的tap分区 consumerGroup 消费者的分组
 *
 * ps: 一个topic下可以有多个tap, *则监听所有分区
 */
@RocketMQMessageListener(topic = "${rocketmq.consumer.topic}"
        , consumerGroup = "${rocketmq.consumer.group}"
        , selectorExpression = "user")
@Component
public class MqTask implements RocketMQListener<MessageExt> {

    /**
     * 消息处理
     *
     * @param messageExt 监听到的消息
     */
    @Override
    public void onMessage(MessageExt messageExt) {
        String topic = messageExt.getTopic();
        String tags = messageExt.getTags();
        byte[] body = messageExt.getBody();
        String keys = messageExt.getKeys();
        String msgId = messageExt.getMsgId();
        String realTopic = messageExt.getProperty("REAL_TOPIC");
        String originMessageId = messageExt.getProperty("ORIGIN_MESSAGE_ID");
        int reconsumeTimes = messageExt.getReconsumeTimes();
        String jsonString = JSON.toJSONString(new String(body));
        System.out.println(String.format("接收到消息:\ntopic【%s】\ntags【%s】\njsonString【%s】" +
                "\nkeys【%s】\nmsgId【%s】\nrealTopic【%s】\noriginMessageId【%s】\nreconsumeTimes【%s】"
                , topic, tags, jsonString, keys, msgId, realTopic, originMessageId, reconsumeTimes));
    }

}
